package com.rabbitmq.rabbitmqextensions.controller;

import com.rabbitmq.rabbitmqextensions.constant.Constants;
import jakarta.annotation.Resource;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.UUID;

@RequestMapping("/producer")
@RestController
public class ProducerController {

    @Resource(name = "normalRabbitTemplate")
    private RabbitTemplate rabbitTemplate;

    @Resource(name = "confirmRabbitTemplate1")
    private RabbitTemplate confirmRabbitTemplate;

    @Resource(name = "transRabbitTemplate")
    private RabbitTemplate transRabbitTemplate;

    @RequestMapping("/ack")
    public String ack() {
        String messageInfo = "consumer ack mode test...";

        rabbitTemplate.convertAndSend(Constants.ACK_EXCHANGE, Constants.ACK_ROUTINGKEY, messageInfo);

        return "消息发送成功";
    }

    /**
     * 持久化消息
     * @return
     */
    @RequestMapping("/pres")
    public String pers() {

        String messageInfo = "pres message";

        Message message = new Message(messageInfo.getBytes(StandardCharsets.UTF_8),
                new MessageProperties());

        //设置消息持久化
        message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);

        rabbitTemplate.convertAndSend(Constants.PRES_EXCHANGE, Constants.PERS_ROUTINGKEY, message);

        return "持久化消息发送成功";
    }

    /**
     * 非持久化
     * @return
     */
    @RequestMapping("/nPres")
    public String nPers() {

        String messageInfo = "nPres message";

        Message message = new Message(messageInfo.getBytes(StandardCharsets.UTF_8),
                new MessageProperties());

        //设置消息非持久化
        message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);

        rabbitTemplate.convertAndSend(Constants.NOT_PERS_EXCHANGE, Constants.PERS_ROUTINGKEY, message);

        return "非持久化消息发送成功";
    }

    /**
     * 确认模式
     */
    @RequestMapping("/confirm")
    public String confirm() {
        String message = "confirm test";

        /**
         * 若在此处设置回调函数，那么会影响到所有 rabbitTemplate
         * 并且只能发送一次消息，因为发送那个多次消息就相当于设置了多个回调函数，规定只能设置一次
         */
//        confirmRabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
//
//            /**
//             *
//             * @param correlationData correlation data for the callback.
//             * @param ack true: 消息到达交换机; false: 消息没有到达交换机
//             * @param cause An optional cause, for nack, when available, otherwise null.
//             */
//            @Override
//            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
//                if (ack) {
//                    System.out.println("接收到消息, 消息 id: " + (null == correlationData ? null : correlationData.getId()));
//                } else {
//                    System.out.println("未接收到消息, 消息 id: " + (null == correlationData ? null : correlationData.getId()) + ", cause: " + cause);
//                }
//            }
//        });

        String id = UUID.randomUUID().toString();

        CorrelationData correlationData = new CorrelationData(id);

        confirmRabbitTemplate.convertAndSend(Constants.CONFIRM_EXCHANGE, Constants.CONFIRM_ROUTINGKEY + "aaa", message, correlationData);

        return "消息发送成功";
    }

    /**
     * 回退模式
     * @return
     */
    @RequestMapping("/returns")
    public String returns() {
        String message = "confirm test";

        String id = UUID.randomUUID().toString();

        CorrelationData correlationData = new CorrelationData(id);

        confirmRabbitTemplate.convertAndSend(Constants.CONFIRM_EXCHANGE, Constants.CONFIRM_ROUTINGKEY + "aaa", message, correlationData);

        return "消息发送成功";
    }

    /**
     * 重试机制
     */
    @RequestMapping("/retry")
    public String retry() {
        String message = "retry test...";

        rabbitTemplate.convertAndSend(Constants.RETRY_EXCHANGE, Constants.RETRY_ROUTINGKEY, message);

        return "消息发送成功";
    }

    /**
     * ttl
     */
    @RequestMapping("/ttl1")
    public String ttl1() {
        String messageInfo = "ttl..." + new Date();

        rabbitTemplate.convertAndSend(Constants.TTL_EXCHANGE, Constants.TTL_ROUTINGKEY, messageInfo, message -> {
            message.getMessageProperties().setExpiration("25000"); //过期时间，单位为 ms
            return message;
        });


        rabbitTemplate.convertAndSend(Constants.TTL_EXCHANGE, Constants.TTL_ROUTINGKEY, messageInfo, message -> {
            message.getMessageProperties().setExpiration("10000"); //过期时间，单位为 ms
            return message;
        });

//        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
//            @Override
//            public Message postProcessMessage(Message message) throws AmqpException {
//                message.getMessageProperties().setExpiration("10000"); //该消息的过期时间为 10s
//                return message;
//            }
//        };

//        rabbitTemplate.convertAndSend(Constants.TTL_EXCHANGE, Constants.TTL_ROUTINGKEY, messageInfo, messagePostProcessor);

        return "消息发送成功";
    }

    @RequestMapping("/ttl2")
    public String ttl2() {

        for (int i = 0; i < 10; i++) {
            String messageInfo = "ttl... " + i;

            rabbitTemplate.convertAndSend(Constants.TTL_EXCHANGE, Constants.TTL_ROUTINGKEY, messageInfo);
        }

        return "消息发送成功";
    }

    @RequestMapping("/dl1")
    public String dl1() {

//        for (int i = 0; i < 10; i++) {
//            String messageInfo = (new Date()).toString() + " " + i;
//            rabbitTemplate.convertAndSend(Constants.NOR_EXCHANGE, Constants.NOR_ROUTINGKEY, messageInfo);
//        }

        String messageInfo = "dl... " + new Date();

        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setExpiration("10000"); //10s 后过期
                return message;
            }
        };

        rabbitTemplate.convertAndSend(Constants.NOR_EXCHANGE, Constants.NOR_ROUTINGKEY, messageInfo, messagePostProcessor);

        return "消息发送成功";
    }

    @RequestMapping("/dl2")
    public String dl2() {

        String messageInfo = "dl ..." + new Date();

        rabbitTemplate.convertAndSend(Constants.NOR_EXCHANGE, Constants.NOR_ROUTINGKEY, messageInfo + " 15000ms", message -> {
            message.getMessageProperties().setExpiration("15000"); //过期时间，单位为 ms
            return message;
        });


        rabbitTemplate.convertAndSend(Constants.NOR_EXCHANGE, Constants.NOR_ROUTINGKEY, messageInfo + " 10000ms", message -> {
            message.getMessageProperties().setExpiration("10000"); //过期时间，单位为 ms
            return message;
        });


        return "消息发送成功";
    }

    /**
     * 延迟
     * @return
     */
    @RequestMapping("/delay")
    public String delay() {

        String messageInfo = "delay ..." + new Date();

        rabbitTemplate.convertAndSend(Constants.DELAY_EXCHANGE, Constants.DELAY_ROUTINGKEY, messageInfo + " 15000ms", message -> {
            message.getMessageProperties().setDelayLong(15000L); //过期时间，单位为 ms
            return message;
        });


        rabbitTemplate.convertAndSend(Constants.DELAY_EXCHANGE, Constants.DELAY_ROUTINGKEY, messageInfo + " 10000ms", message -> {
            message.getMessageProperties().setDelayLong(10000L); //过期时间，单位为 ms
            return message;
        });

        return "消息发送成功";
    }

    /**
     * 事务
     */
    @Transactional //开启事务
    @RequestMapping("/trans")
    public String trans() {
        transRabbitTemplate.convertAndSend("", Constants.TRANS_QUEUE, "trans 1");
        transRabbitTemplate.convertAndSend("", Constants.TRANS_QUEUE, "trans 2");
        int num = 3 / 0;
        transRabbitTemplate.convertAndSend("", Constants.TRANS_QUEUE, "trans 3");

        return "消息发送成功";
    }

    /**
     * 限流
     */
    @RequestMapping("/qos")
    public String qos() {
        for (int i = 0; i < 20; i++) {

            String messageInfo = "qos... " + i;

            rabbitTemplate.convertAndSend(Constants.QOS_EXCHANGE, Constants.QOS_ROUTINGKEY, messageInfo);
        }

        return "消息发送成功";
    }
}
